package com.booter.zookeeper;


import com.booter.zookeeper.exception.ZkException;
import com.booter.zookeeper.listener.ChildListener;
import com.booter.zookeeper.listener.ConnStateListener;
import com.booter.zookeeper.listener.DataListener;
import com.booter.zookeeper.listener.StateListener;
import com.booter.zookeeper.serializer.SerializableSerializer;
import com.booter.zookeeper.serializer.ZkSerializer;
import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author admin
 */
public class CuratorZkClient extends
        AbstractZkClient<CuratorZkClient.PathChildrenListener, CuratorZkClient.NodeListener> {

    private static final Logger LOGGER = LoggerFactory.getLogger(CuratorZkClient.class);

    private final CuratorFramework client;

    private final ZkSerializer zkSerializer;

    private final CuratorConfig config;

    private final Set<EphemeralNode> ephemeralNodes = new CopyOnWriteArraySet<>();

    private final List<ACL> acls = new ArrayList<>();

    private String getDigest() {
        return this.config.getAuth().concat(":").concat(this.config.getPassword());
    }


    public CuratorZkClient(CuratorConfig config) {
        this.config = config;
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory
                .builder().connectString(config.getHost())
                .namespace(config.getNameSpace())
                .retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
                .sessionTimeoutMs(config.getSessionTimeout())
                .connectionTimeoutMs(config.getConnectionTimeout())
                .maxCloseWaitMs(1000);

        if (config.getAuth() != null && !config.getAuth().isEmpty()) {
            builder.aclProvider(new ACLProvider() {
                private List<ACL> acl;

                @Override
                public List<ACL> getDefaultAcl() {
                    if (this.acl == null) {
                        List<ACL> acl = ZooDefs.Ids.CREATOR_ALL_ACL;
                        acl.clear();
                        acl.add(new ACL(ZooDefs.Perms.ALL, new Id("auth", CuratorZkClient.this.getDigest())));
                        this.acl = acl;
                    }
                    return this.acl;
                }

                @Override
                public List<ACL> getAclForPath(String path) {
                    return this.acl;
                }
            });
            builder.authorization("digest", config.getAuth().getBytes());
            this.acls.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", this.getDigest())));
        }
        this.client = builder.build();

        this.client.getConnectionStateListenable().addListener((c, state) -> {
            if (state == ConnectionState.LOST) {
                LOGGER.warn("THIS HOST IS DISCONNECTED ");
                this.stateChanged(ConnStateListener.DISCONNECTED);
            } else if (state == ConnectionState.CONNECTED) {
                LOGGER.info("THIS HOST IS CONNECTED ");
                this.stateChanged(ConnStateListener.CONNECTED);
            } else if (state == ConnectionState.RECONNECTED) {
                LOGGER.info("THIS HOST IS RECONNECTED ");
                this.stateChanged(ConnStateListener.RECONNECTED);
                //重建临时节点
                ephemeralNodes.forEach((k) -> {
                    if (k != null) {
                        if (k.getData() == null) {
                            this.createEphemeral(k.getKey(), k.isSequential());
                        } else {
                            this.createEphemeral(k.getKey(), k.getData(), k.isSequential());
                        }
                    }
                });
            } else if (state == ConnectionState.SUSPENDED) {
                LOGGER.warn("THIS HOST IS SUSPENDED ");
                this.stateChanged(ConnStateListener.DISCONNECTED);
            }
        });
        this.zkSerializer = new SerializableSerializer();
        this.client.start();
    }

    public CuratorFrameworkState getState() {
        return this.client.getState();
    }

    @Override
    protected String createPersistent(String path, boolean sequential) {
        try {
            if (sequential) {
                return this.client.create()
                        .withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(path);
            } else {
                return this.client.create().withMode(CreateMode.PERSISTENT)
                        .forPath(path);
            }
        } catch (KeeperException.NodeExistsException e) {
            LOGGER.debug("节点已经创建,{}", e.getMessage());
            return path;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected String createPersistent(String path, Object data, boolean sequential) {
        try {
            byte[] bytes;
            if (data instanceof Serializable) {
                bytes = this.zkSerializer.serialize(data);
            } else {
                bytes = (byte[]) data;
            }
            if (sequential) {
                return this.client.create()
                        .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                        .forPath(path, bytes);
            } else {
                return this.client.create().withMode(CreateMode.PERSISTENT)
                        .forPath(path, bytes);
            }
        } catch (KeeperException.NodeExistsException e) {
            LOGGER.debug("节点已经创建,{}", e.getMessage());
            return path;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected String createEphemeral(String path, boolean sequential) {
        EphemeralNode node = new EphemeralNode(path, null, sequential);
        try {
            ephemeralNodes.add(node);
            if (sequential) {
                return this.client.create()
                        .withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
            } else {
                return this.client.create().withMode(CreateMode.EPHEMERAL)
                        .forPath(path);
            }
        } catch (KeeperException.NodeExistsException e) {
            LOGGER.debug("节点已经创建,{}", e.getMessage());
            return path;
        } catch (Exception e) {
            ephemeralNodes.remove(node);
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected String createEphemeral(String path, Object data,
                                     boolean sequential) {
        EphemeralNode node = new EphemeralNode(path, data, sequential);
        try {
            ephemeralNodes.add(node);
            byte[] bytes;
            if (data instanceof Serializable) {
                bytes = this.zkSerializer.serialize(data);
            } else {
                bytes = (byte[]) data;
            }
            if (sequential) {
                return this.client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                        .forPath(path, bytes);
            } else {
                return this.client.create().withMode(CreateMode.EPHEMERAL).forPath(path, bytes);
            }
        } catch (KeeperException.NodeExistsException e) {
            LOGGER.debug("节点已经创建,{}", e.getMessage());
            return path;
        } catch (Exception e) {
            ephemeralNodes.remove(node);
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected PathChildrenListener createTargetChildListener(String path, ChildListener listener) {
        return new PathChildrenListener(path, listener);
    }

    @Override
    protected List<String> addTargetChildListener(String path, PathChildrenListener listener) {
        try {
            listener.startListener();
            return this.client.getChildren().forPath(path);
        } catch (KeeperException.NoNodeException e) {
            LOGGER.warn("node is not exist,ex:{}", e.getMessage());
            return null;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected void removeTargetChildListener(String path, PathChildrenListener listener) {
        listener.stopListener();
    }

    @Override
    protected void addTargetDataListener(String path, NodeListener listener) {
        listener.startListener();
    }

    @Override
    protected NodeListener createTargetDataListener(String path, final DataListener listener) {
        return new NodeListener(path, listener);
    }

    @Override
    protected void removeTargetDataListener(String path, NodeListener listener) {
        listener.stopListener();
    }

    @Override
    public boolean delete(String path) {
        try {
            this.client.delete().forPath(path);
            return true;
        } catch (KeeperException.NoNodeException e) {
            LOGGER.warn("node is not exist,ex:{}", e.getMessage());
            return true;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public boolean exists(String path) {
        try {
            return this.client.checkExists().forPath(path) != null;
        } catch (Exception e) {//NPM
            LOGGER.warn("检查节点是否存在,ex:{}", e.getMessage());
            return false;
        }
    }

    @Override
    public <T> T getData(String path) {
        try {
            return (T) this.zkSerializer.deserialize(this.client.getData().forPath(path));
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public void setData(String path, Object data) {
        byte[] bytes;
        if (data instanceof Serializable) {
            bytes = this.zkSerializer.serialize(data);
        } else {
            bytes = (byte[]) data;
        }
        try {
            this.client.setData().forPath(path, bytes);
            LOGGER.info("set data succeed, path:{}, data:{}", path, data);
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public List<String> getChildren(String path) {
        try {
            return this.client.getChildren().forPath(path);
        } catch (KeeperException.NoNodeException e) {
            LOGGER.warn("node is not exist，ex:{}", e.getMessage());
            return null;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public boolean isConnected() {
        return this.client.getZookeeperClient().isConnected();
    }

    @Override
    protected void doClose() {
        this.client.close();
    }

    public class PathChildrenListener {
        private final PathChildrenCache childrenCache;
        private final PathChildrenCacheListener childrenCacheListener;
        private final AtomicBoolean start = new AtomicBoolean(false);

        public PathChildrenListener(String path, final ChildListener listener) {
            this.childrenCache = new PathChildrenCache(CuratorZkClient.this.client, path, true);
            this.childrenCacheListener = (c, event) -> {
                if (event.getData() == null) {
                    return;
                }
                String childPath = event.getData().getPath();
                String parentPath = childPath.substring(0, childPath.lastIndexOf('/'));
                List<String> children = CuratorZkClient.this.client.getChildren().forPath(parentPath);

                switch (event.getType()) {
                    case CHILD_ADDED:
                        listener.childChanged(parentPath, children,
                                PathChildrenCacheEvent.Type.CHILD_ADDED.name());
                        break;
                    case CHILD_REMOVED:
                        listener.childChanged(parentPath, children,
                                PathChildrenCacheEvent.Type.CHILD_REMOVED.name());
                        break;
                    case CHILD_UPDATED:
                        listener.childChanged(parentPath, children,
                                PathChildrenCacheEvent.Type.CHILD_UPDATED.name());

                        break;
                    default:
                        break;
                }
            };
        }

        public void startListener() {
            try {
                if (this.start.compareAndSet(false, true)) {
                    this.childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
                    this.childrenCache.getListenable().addListener(this.childrenCacheListener);
                }
            } catch (Exception e) {
                throw new ZkException(e);
            }
        }

        public void stopListener() {
            try {
                if (this.start.compareAndSet(true, false)) {
                    this.childrenCache.getListenable().removeListener(this.childrenCacheListener);
                    this.childrenCache.clear();
                    this.childrenCache.close();
                }
            } catch (IOException e) {
                throw new ZkException(e);
            }
        }
    }

    public class NodeListener {
        private final NodeCache nodeCache;
        private final NodeCacheListener nodeCacheListener;
        private final AtomicBoolean start = new AtomicBoolean(false);

        public NodeListener(String path, final DataListener listener) {
            this.nodeCache = new NodeCache(CuratorZkClient.this.client, path, false);
            this.nodeCacheListener = () -> {
                String path1 = this.nodeCache.getCurrentData().getPath();
                byte[] data = this.nodeCache.getCurrentData().getData();
                if (data == null) {
                    listener.dataDeleted(path1);
                } else {
                    listener.dataChange(path1, data);
                }
            };
        }

        public void startListener() {
            try {
                if (this.start.compareAndSet(false, true)) {
                    this.nodeCache.start(true);
                    this.nodeCache.getListenable().addListener(this.nodeCacheListener);
                }
            } catch (Exception e) {
                throw new ZkException(e);
            }
        }

        public void stopListener() {
            try {
                if (this.start.compareAndSet(true, false)) {
                    this.nodeCache.getListenable().removeListener(this.nodeCacheListener);
                    this.nodeCache.close();
                }
            } catch (IOException e) {
                throw new ZkException(e);
            }
        }
    }

    public CuratorFramework getClient() {
        return this.client;
    }

    public CuratorConfig getConfig() {
        return this.config;
    }

//    public void addStateListener(StateListener listener) {
//        if (listener == null) {
//            return;
//        }
//        LOGGER.info("Current Node Id is :{}", listener.getId());
//        try (LeaderLatch latch = new LeaderLatch(this.getClient(),
//            listener.getLatchPath(), listener.getId());) {
//            latch.addListener(new LeaderSelecterListener(listener, latch));
//            latch.start();
//        } catch (Exception e) {
//            LOGGER.error("", e);
//        }
//    }

    public void addStateListener(StateListener listener) {
        if (listener == null) {
            return;
        }
        LOGGER.info("Current Node Id is :{}", listener.getId());
        try (LeaderLatch latch = new LeaderLatch(this.getClient(),
                listener.getLatchPath(), listener.getId())) {
            latch.addListener(new LeaderSelectorListener(listener, latch));
            latch.start();
        } catch (Exception e) {
//            try {
//                if (latch != null) {
//                    latch.close();
//                }
//            } catch (final IOException ioe) {
//                LOGGER.error("", ioe);
//            }
            LOGGER.error("", e);
        }
    }

    private class LeaderSelectorListener implements LeaderLatchListener {
        private final StateListener listener;
        private final LeaderLatch latch;

        public LeaderSelectorListener(final StateListener listener, final LeaderLatch latch) {
            this.listener = listener;
            this.latch = latch;
        }

        @Override
        public void isLeader() {
            LOGGER.info("Current Node to Leader:{}", this.latch.hasLeadership());
            if (this.latch.hasLeadership()) {
                this.listener.toLeader();
            } else {
                this.listener.toFollow();
            }
        }

        @Override
        public void notLeader() {
            LOGGER.info("Current Node to Follow:{}", this.latch.hasLeadership());
            if (!this.latch.hasLeadership()) {
                this.listener.toFollow();
            } else {
                this.listener.toLeader();
            }
        }
    }


    private class EphemeralNode {
        private String key;
        private Object data;
        private boolean sequential;


        public EphemeralNode(String key, Object data, boolean sequential) {
            this.key = key;
            this.data = data;
            this.sequential = sequential;
        }

        public String getKey() {
            return key;
        }

        public void setKey(String key) {
            this.key = key;
        }

        public Object getData() {
            return data;
        }

        public void setData(Object data) {
            this.data = data;
        }

        public boolean isSequential() {
            return sequential;
        }

        public void setSequential(boolean sequential) {
            this.sequential = sequential;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }

            if (o == null || getClass() != o.getClass()) {
                return false;
            }

            EphemeralNode that = (EphemeralNode) o;
            return new EqualsBuilder().append(sequential, that.sequential).append(key, that.key).append(data, that.data).isEquals();
        }

        @Override
        public int hashCode() {
            return new HashCodeBuilder(17, 37).append(key).append(data).append(sequential).toHashCode();
        }
    }


}
